package sjc;

import com.google.common.collect.Lists;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.shared.SharedCount;
import org.apache.curator.framework.recipes.shared.SharedCountListener;
import org.apache.curator.framework.recipes.shared.SharedCountReader;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 如果真的要实现分布式计数器，应该是用redis来实现，redis的并发量更高，性能更好，功能更加的强大。
 *
 * SharedCount计数器 使用int类型来计数。 主要涉及三个类。
 *      - SharedCount : 管理一个共享的整数。所有看同样的路径客户端将有共享的整数（考虑ZK的正常一致性保证）的最高最新的值。
 *      - SharedCountReader : 一个共享的整数接口，并允许监听改变它的值。
 *      - SharedCountListener : 用于监听共享整数发生变化的监听器。
 */
public class SharedCounterTest implements SharedCountListener {
    private static final String PATH = "/examples/counter";

    public static void main(String[] args) throws Exception {
        final Random random = new Random();
        SharedCounterTest example = new SharedCounterTest();

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3, Integer.MAX_VALUE);
        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.72.133:2181", 3000, 3000, retryPolicy);
        client.start();

        SharedCount baseCount = new SharedCount(client, PATH, 0);
        baseCount.addListener(example);
        baseCount.start();

        ArrayList<SharedCount> examples = Lists.newArrayList();
        ExecutorService service = Executors.newFixedThreadPool(5);

        for(int i=0; i<5; i++){
            final SharedCount count = new SharedCount(client, PATH, 0);
            examples.add(count);

            Callable<Void> task = new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    count.start();
                    Thread.sleep(random.nextInt(10000));
                    count.setCount(random.nextInt(10000)); // 强制设置值
                    System.out.println("计数器当前值：" + count.getVersionedValue().getValue());
                    System.out.println("计数器当前版本：" + count.getVersionedValue().getVersion());
                    // 尝试设置值
                    System.out.println("trySetCount:" + count.trySetCount(count.getVersionedValue(), 123));
                    return null;
                }
            };
            service.submit(task);
        }

        service.shutdown();
        service.awaitTermination(1, TimeUnit.MINUTES);

        for(int i=0; i<5; i++){
            examples.get(i).close();
        }
        baseCount.close();
        client.close();
        System.out.println("ok");
    }

    @Override
    public void stateChanged(CuratorFramework client, ConnectionState newState) {
        System.out.println("连接状态 ===> " +  newState.toString());
    }

    @Override
    public void countHasChanged(SharedCountReader sharedCount, int newCount) throws Exception {
        System.out.println("计数器改变 ==> " + newCount);
    }
}
